package com.jida.hadoop.mr.join;

 
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
 
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
 
public class ReducerDxProMatch extends Reducer<Text,Text,Text,NullWritable>{
    Text k3 = new Text();
 
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        List<String> l1 = new ArrayList<String>();
        List<String> l2 = new ArrayList<String>();
        for(Text v:values){
            int i = v.toString().split("\\|").length;
            if(i==1){//用户ID
                l1.add(v.toString());
            }else {//产品信息
                l2.add(v.toString());
            }
        }
        //取出数据
        for(String s1:l1){
            for (String s2:l2){
                k3.set(s1+s2+"|"+key);//用户ID|产品信息...|行为ID
                context.write(k3,NullWritable.get());
            }
        }
    }
}
